package com.personal.gateway.service.impl;

import com.alibaba.fastjson.JSON;
import com.personal.gateway.listener.GatewayRouteListener;
import com.personal.gateway.custom.customEnum.FlinkGateWayCommandEnum;
import com.personal.gateway.dto.ContainerGroup;
import com.personal.gateway.entity.FlinkCommandHistory;
import com.personal.gateway.entity.FlinkGatewayRoute;
import com.personal.gateway.mapper.FlinkCommandHistoryMapper;
import com.personal.gateway.mapper.FlinkGatewayRouteMapper;
import com.personal.gateway.service.FlinkConfigService;
import com.personal.gateway.service.K8sService;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @program: FlinkGateway
 * @description:
 * @author: yangliao
 * @create: 2021-08-23
 */
@Slf4j
@Service
@AllArgsConstructor
public class FlinkConfigServiceImpl implements FlinkConfigService {

    private final K8sService k8sService;

    private final DynamicRouteService dynamicRouteService;

    private final FlinkGatewayRouteMapper flinkGatewayRouteMapper;

    private final FlinkCommandHistoryMapper flinkCommandHistoryMapper;

    private static final ThreadPoolExecutor executor = new ThreadPoolExecutor(
            5, 10, 200, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(5));

    @Override
    public String edit(FlinkGatewayRoute route) throws ExecutionException, InterruptedException {
        // 1、判断添加还是修改
        boolean save = null == route.getId() || route.getId() < 1;

        // 2、数据校验
        String check = checkRouteData(route, save);
        if (StringUtils.isNoneBlank(check)) {
            return check;
        }

        // 3、数据入库
        if (save) {
            route.setGatewayNum(UUID.randomUUID().toString());
            route.setAlarmStatus(0);
            route.setTestingInterval(10);
            route.setAlarmEmail("");
            flinkGatewayRouteMapper.insert(route);

            FlinkCommandHistory history = new FlinkCommandHistory(route.getGatewayNum(), FlinkGateWayCommandEnum.CREATE.getValue(), route.getUserName());
            flinkCommandHistoryMapper.insert(history);
        } else {
            FlinkGatewayRoute bak = flinkGatewayRouteMapper.getRouteById(route.getId());

            // 修改先删除后插入
            flinkGatewayRouteMapper.delete(route.getUserName(), route.getId());

            route.setAlarmStatus(bak.getAlarmStatus());
            route.setTestingInterval(bak.getTestingInterval());
            route.setAlarmEmail(bak.getAlarmEmail());
            flinkGatewayRouteMapper.insert(route);

            String deleteRouteReq = dynamicRouteService.delete(route.getRouteId());
            log.info("删除路由返回结果 {} {}", route.getRouteId(), deleteRouteReq);

            FlinkCommandHistory history = new FlinkCommandHistory(route.getGatewayNum(), FlinkGateWayCommandEnum.EDIT.getValue(), route.getUserName());
            flinkCommandHistoryMapper.insert(history);
        }

        // 4、更新路由缓存 添加路由
        String addRouteReq = dynamicRouteService.add(GatewayRouteListener.initRouteDefinition(route));
        log.info("添加路由返回结果 {} {}", route.getRouteId(), addRouteReq);

        return "success";
    }

    @Override
    public String alarmEdit(FlinkGatewayRoute route) {
        flinkGatewayRouteMapper.updateAlarm(route);
        FlinkCommandHistory history = new FlinkCommandHistory(route.getGatewayNum(), FlinkGateWayCommandEnum.ALARM_EDIT.getValue(), route.getUserName());
        flinkCommandHistoryMapper.insert(history);
        return "success";
    }

    @Override
    public FlinkGatewayRoute getProjectById(Integer id) {
        return flinkGatewayRouteMapper.getRouteById(id);
    }

    @Override
    public String del(Integer id, String user) {

        FlinkGatewayRoute route = flinkGatewayRouteMapper.getRouteById(id);

        // 先判断服务有没有启动 如果没有启动标记删除 在更新路由
        List<ContainerGroup> containerGroups = k8sService.getContainerGroup();
        List<ContainerGroup> jobManager = containerGroups.stream()
                .filter(x -> x.getComponent().equals("jobmanager") && x.getApp().equals(route.getRouteId()))
                .collect(Collectors.toList());
        int jobManagerSize = jobManager.size();

        if (jobManagerSize > 0) {
            return "请先停止服务后在操作删除！";
        }

        flinkGatewayRouteMapper.delete(user, id);

        String deleteRouteReq = dynamicRouteService.delete(route.getRouteId());
        log.info("删除路由返回结果 {} {}", route.getRouteId(), deleteRouteReq);

        FlinkCommandHistory history = new FlinkCommandHistory(route.getGatewayNum(), FlinkGateWayCommandEnum.DELETE.getValue(), route.getUserName());
        flinkCommandHistoryMapper.insert(history);

        return "success";
    }

    @Override
    public String routeView() throws ExecutionException, InterruptedException {
        return JSON.toJSONString(new ArrayList<>(executor.submit(() -> dynamicRouteService.routes().block()).get()));
    }

    @Override
    public List<FlinkCommandHistory> history(Integer id) {
        FlinkGatewayRoute route = flinkGatewayRouteMapper.getRouteById(id);
        return flinkCommandHistoryMapper.list(route.getGatewayNum());
    }

    @Override
    public FlinkCommandHistory getHistoryById(Integer id) {
        return flinkCommandHistoryMapper.getHistoryById(id);
    }

    private String checkRouteData(FlinkGatewayRoute route, boolean save) {
        if (StringUtils.isBlank(route.getRouteId())) {
            return "请输入路由ID";
        }
        if (StringUtils.isBlank(route.getRouteUri())) {
            return "请输入路由地址";
        }
        if (StringUtils.isBlank(route.getRoutePredicates())) {
            return "请输入匹配路径";
        }
        if (null == route.getRouteFiltersStripprefix()) {
            return "请选择过滤层级";
        }

        if (StringUtils.isBlank(route.getProjectSummary())) {
            return "请输入需求简介";
        }
        if (StringUtils.isBlank(route.getProjectDetail())) {
            return "请输入访问地址";
        }

        if (StringUtils.isBlank(route.getFlinkVersion())) {
            return "请先择Flink版本";
        }
        if (StringUtils.isBlank(route.getStartClass())) {
            return "请输入启动路径";
        }
        if (StringUtils.isBlank(route.getTaskManagerMemory())) {
            return "请输入启动内存";
        }
        if (StringUtils.isBlank(route.getStartJarName())) {
            return "请输入JAR名称";
        }
        if (StringUtils.isBlank(route.getImageName())) {
            return "请输入镜像地址";
        }

        // 新增校验 数据库是否存在
        if (save) {
            if (flinkGatewayRouteMapper.countByRouteId(route.getRouteId()) > 0) {
                return "路由ID已经存在";
            }
            if (flinkGatewayRouteMapper.countByRouteUri(route.getRouteUri()) > 0) {
                return "匹配路径已经存在";
            }
        } else {
            if (StringUtils.isBlank(route.getGatewayNum())) {
                return "项目唯一编号错误";
            }
        }

        return "";
    }

}
